package com.seari.client.kafka;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import com.seari.main.StartupClass;

@Component
public class KafkaStarter
{
	protected static Logger logger = LoggerFactory.getLogger(KafkaStarter.class);
	public int sliceCount; //每个线程所包含的主题数
	public HashMap<String, KafkaWorker> kafkaWorkerRefMap = new HashMap<>();
	//初始化kafka启动器
	public void init()
	{
		try
		{
			//获取订阅的主题列表
			List<String> topicList = Arrays.asList(StartupClass.config.getProperty("kafka_topics").split(","));
			sliceCount = Integer.parseInt(StartupClass.config.getProperty("kafka_topics_slice_count"));
			if(!CollectionUtils.isEmpty(topicList))
			{
				int num1 = topicList.size() / sliceCount;
				int num2 = topicList.size() % sliceCount;
				if(topicList.size() >= sliceCount)
				{
					for(int i = 0; i < num1; i++)
					{
						List<String> topics = topicList.subList(i * sliceCount, (i * sliceCount) + sliceCount);
						String threadName = "kafka worker - " + i;
						KafkaWorker kafkaWorker = new KafkaWorker(topics,threadName);
						kafkaWorkerRefMap.put(threadName, kafkaWorker);
						new Thread(kafkaWorker).start();
						logger.info("create kafka worker:{},topics size:{}",threadName,topics.size());
					}
				}
				if(num2 > 0)
				{
					List<String> topics = topicList.subList(num1 * sliceCount, topicList.size());
					String threadName = "kafka worker rest";
					KafkaWorker kafkaWorker = new KafkaWorker(topics, threadName);
					kafkaWorkerRefMap.put(threadName, kafkaWorker);
					new Thread(kafkaWorker).start();
					logger.info("create kafka worker:{},topics size:{}",threadName,topics.size());
				}
			}
		} catch (Exception e)
		{
			e.printStackTrace();
		}
	}
}
